package com.yc.myspringboot;

import com.rabbitmq.client.*;
import org.junit.Test;
import com.yc.myspringboot.mq.*;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * description:
 * author: yc
 * date: 2021/4/22 10:19
 */
public class mq extends BaseTest {

//    @Autowired
//    private Creater11 creater;

    @Test
    public void testMqProducer() {
        try {
            // 创建连接工厂，并进行属性设置
            ConnectionFactory connectionFactory = new ConnectionFactory();
//            connectionFactory.setHost("10.1.32.114");
//            connectionFactory.setPort(5672);
//            connectionFactory.setVirtualHost("/");

            // 第二种方式：amqp://用户名:密码@IP:端口号/虚拟主机名称 (有其他主机名称时应该能使用)
		    connectionFactory.setUri("amqp://java:Cdd20180406@10.1.32.99:5672/");
            // 获取连接
            Connection connection = connectionFactory.newConnection();
            // 创建信道
            Channel channel = connection.createChannel();

            // 通过信道发送信息
            String msg = "hello, rabbitmq";
//		channel.basicPublish(exchange, routingKey, props, body);
            // 这里没有设置交换机，routingKey 必须设置成跟队列名一样，队列才能收到消息
            channel.basicPublish("", "test.queue", null, msg.getBytes());
            System.out.println("生产端消息：" + msg);
            // 关闭连接
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static final String EXCHANGE_NAME = "exchange_demo";
    public static final String ROUTING_KEY = "routingkey_demo";
    public static final String QUEUE_NAME = "queue_demo";
    public static final String IP_ADDRESS = "10.1.32.99";
    public static final int PORT = 5672;


    @Test
    public void testMqProducer1() {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("java");
            factory.setPassword("Cdd20180406");
            connection = factory.newConnection();	// 建立连接
            channel = connection.createChannel();// 创建信道
            // 创建一个type="direct"、持久化的、非自动删除的交换器
            //            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
            //            // 创建一个持久化、非排他的、非自动删除的队列
            //            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //            // 将交换器和队列通过路由绑定
            //            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 发送一条持久化的消息：hello world!

            for (int j =0; j<100; j++) {
                String message = Thread.currentThread().getName() + j;
                channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        message.getBytes());
                System.out.println("写入：" + message);
                Thread.sleep(100);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            try {
                channel.close();
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }

        }


    }


    @Test
    public void testMqProducer2() {

        for (int i=0; i<10; i++) {//10个线程，每个线程写100次, 每次间隔100毫秒
            if (i<5) {
                new Thread(new ProducerTask()).start();
            } else {
                new Thread(new ConsumerTask()).start();
            }

        }

        try {
            Thread.sleep(15000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    class ProducerTask implements Runnable {

        @Override
        public void run() {
            Connection connection = null;
            Channel channel = null;
            try {
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost(IP_ADDRESS);
                factory.setPort(PORT);
                factory.setUsername("java");
                factory.setPassword("Cdd20180406");
                connection = factory.newConnection();	// 建立连接
                channel = connection.createChannel();// 创建信道
                // 创建一个type="direct"、持久化的、非自动删除的交换器
                //            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
                //            // 创建一个持久化、非排他的、非自动删除的队列
                //            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                //            // 将交换器和队列通过路由绑定
                //            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
                // 发送一条持久化的消息：hello world!

                for (int j =0; j<100; j++) {
                    String message = Thread.currentThread().getName() + j;
                    channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBytes());
                    System.out.println("写入：" + message);
                    Thread.sleep(100);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 关闭资源
                try {
                    channel.close();
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            }
        }
    }

    class ConsumerTask implements Runnable {

        @Override
        public void run() {
            try {
                Address[] addresses = new Address[] {
                        new Address(IP_ADDRESS, PORT)
                };
                ConnectionFactory factory = new ConnectionFactory();
                factory.setUsername("java");
                factory.setPassword("Cdd20180406");
                // 这里的连接方式与生产者的demo略有不同，注意区分
                Connection connection = factory.newConnection(addresses);	// 创建连接
                final Channel channel = connection.createChannel();	// 创建信道
                channel.basicQos(64); 	// 设置客户端最多接受未被ack的消息的个数
                com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        System.out.println("recv message: " + new String(body));
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                };
                channel.basicConsume(QUEUE_NAME, consumer);
                // 等待回调函数执行完毕后，关闭资源
                TimeUnit.SECONDS.sleep(5);
                channel.close();
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    @Test
    public void testRabbitConsumer() {
        String EXCHANGE_NAME = "exchange_demo";
        String ROUTING_KEY = "routingkey_demo";
        String QUEUE_NAME = "queue_demo";
        String IP_ADDRESS = "10.1.32.99";
        int PORT = 5672;	// RabbitMQ服务端默认端口号为5672

        try {
            Address[] addresses = new Address[] {
                    new Address(IP_ADDRESS, PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("java");
            factory.setPassword("Cdd20180406");
            // 这里的连接方式与生产者的demo略有不同，注意区分
            Connection connection = factory.newConnection(addresses);	// 创建连接
            final Channel channel = connection.createChannel();	// 创建信道
            channel.basicQos(64); 	// 设置客户端最多接受未被ack的消息的个数
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("recv message: " + new String(body));
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
            // 等待回调函数执行完毕后，关闭资源
            TimeUnit.SECONDS.sleep(500);
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }




//    @Test
//    public void creatertest(){
//        creater.create();
//    }

}
